//
// Created by abstergo on 24-11-5.
//
#include <base/ThreadPool.h>
#include <base/log.hpp>
#include <vector>
#include <MessageQueue/MessageQueue.h>
#include <base/lock.hpp>
#undef INUSE_LOGGERS

#undef LOG_LEVEL

#define LOG_LEVEL L_DEBUG

#include <base/loggers/BuiltInLogger.h>

BuiltInLoggers::ConsoleLogger logger("threadPool-logger");

#define INUSE_LOGGERS &logger

void logReject(const char *threadName, CONST Task *task, const char *reason) {

  WARN_LOG("Task: {},reject by thread:{}", task->taskId, threadName);

}
Task createTask(task_function task, unsigned long long int taskId, void *data, unsigned long dataSize) {
  return Task(task, data, dataSize, taskId);
}

struct PoolItem {
  MessageQueue::MsgQueue *_queue;
  std::thread *_thread;
  bool _is_running = false;
  std::string _thread_name;
  ThreadPoolConfig _config;
  virtual ~PoolItem() {
	_is_running = false;
	if (_queue != nullptr) {
	  _queue->cancel();
	  delete _queue;
	}
	if (_thread != nullptr) {
	  delete _thread;
	  _thread = nullptr;
	}
	INFO_LOG("Thread:{} exit", _thread_name);

  }
  PoolItem(CONST PoolItem &) = delete;
  PoolItem() = default;
};

using lock_type = tbs::lock::WriteReadLock<tbs::lock::MutexLock>;
lock_type _lock;

std::vector<PoolItem *> _pool;
int _index = 0;
reject_function _reject;

void useThreadPool(std::initializer_list<ThreadPoolConfig> configs, reject_function r) {
  _reject = r;
  for (auto &config : configs) {
	PoolItem *item = new PoolItem();
	item->_queue = new MessageQueue::MsgQueue();
	item->_thread_name = config.threadName;
	item->_thread = new std::thread([item]() {
	  INFO_LOG("Thread:{} start", item->_thread_name);
	  item->_is_running = true;
	  item->_queue->execute();
	  item->_is_running = false;
	  INFO_LOG("Thread:{} End", item->_thread_name);

	});
	item->_config = config;
	_pool.push_back(item);
	item->_thread->detach();
  }

}
Task *submitTask(const Task &t) {
  PoolItem *target = nullptr;
  {
	tbs::lock::make_read_lock_guard(_lock);
	int index = (++_index) % _pool.size();
	if (_pool[index]->_config.maxTaskCount != -1
		&& _pool[index]->_queue->messageCount() >= _pool[index]->_config.maxTaskCount) {
	  for (index = 0; index < _pool.size(); index++) {
		if (_pool[index]->_queue->messageCount() < _pool[index]->_config.maxTaskCount) {
		  break;
		}
	  }
	}
	if (index >= _pool.size()) {
	  _reject("null thread", &t, "No available thread");
	  return nullptr;
	}
	target = _pool[index];
  }
  Task *rt = new Task(t);

  MessageQueue::DelayMessage msg;
  msg.setConsumeMethod([target](MessageQueue::DelayMessage &m) {
	Task *ta = m.seek<Task *>(0);
	ta->status = TaskStatus_RUNNING;
	try {
	  ta->task(ta->data, ta->dataSize, target->_thread_name.c_str(), ta->taskId);
	  ta->status = TaskStatus_FINISH;
	} catch (std::exception &e) {
	  ta->status = TaskStatus_Error;
	  _reject(target->_thread_name.c_str(), ta, e.what());
	}
	delete ta;
	ta = nullptr;
	return MessageQueue::DelayMessage::StatusEnum::Consumed;
  }).append_t(rt);
  target->_queue->send(std::move(msg));
  return rt;
}
Task::Task(const task_function &Task, void *Data, unsigned long DataSize, unsigned long long int TaskId)
	: task(Task), data(Data), dataSize(DataSize), taskId(TaskId) {}
